package com.spring.component.rabbitmq;

import java.io.IOException;
import java.util.concurrent.TimeoutException;
import java.util.stream.Stream;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.spring.component.utils.CollectionUtil;
import com.spring.component.utils.StringUtil;

/**
 * @author liushihua
 * @since 2018年04月11日
 */
@Component
public class QueueBuilder {
  private final Logger log = LoggerFactory.getLogger(this.getClass());

  @Resource
  private CachingConnectionFactory connectionFactory;
  @Resource(name = "iotbusConnectionFactory")
  private CachingConnectionFactory iotBusConnectionFactory;
  @Resource(name = "logserviceConnectionFactory")
  private CachingConnectionFactory logserviceConnectionFactory;
  @Autowired
  private Properties rabbitProperties;
  @Autowired
  private TopicExchange topicExchange;
  @Autowired
  private TopicExchange iotbusTopicExchange;

  @Bean
  public QueueBuilder intiQueue() {
    log.info("系统加载定义queue......");
    // 初始化默认mq队列
    if (rabbitProperties.validQueue()) {
      // 声明队列
      queueDeclare(connectionFactory, rabbitProperties.getQueues());
      // 声明订阅topic的队列
      queueDeclare(connectionFactory, rabbitProperties.getTopics());
    }
    // 初始化iotbus mq队列
    if (rabbitProperties.validIotbusQueue()) {
      queueDeclare(iotBusConnectionFactory, rabbitProperties.getIotbusQueues());
      queueDeclare(iotBusConnectionFactory, rabbitProperties.getIotbusTopics());
    }
    // 初始化logservice mq队列
    if (rabbitProperties.validLogQueue()) {
      queueDeclare(logserviceConnectionFactory, rabbitProperties.getLogQueues());
    }
    // 绑定默认mq topic routingKey到指定交换机上
    bindExchange(connectionFactory, topicExchange.getName(), rabbitProperties.getRouting());
    // 绑定iotbus mq topic routingKey
    bindExchange(iotBusConnectionFactory, iotbusTopicExchange.getName(),
        rabbitProperties.getIotbusRouting());
    return new QueueBuilder();
  }

  /**
   * 声明队列,根据队列是否配置auotAck自动应答创建队列
   * 
   * @param connectionFactory
   * @param queues
   */
  private void queueDeclare(CachingConnectionFactory connectionFactory, String queues) {
    Channel channel = channel(connectionFactory);
    if (null == channel) {
      log.error("channel为空，queueDeclare失败");
      return;
    }
    if (StringUtil.isNotBlank(queues)) {
      Stream.of(queues.trim().split(",")).filter(StringUtil::isNotBlank).forEach(queue -> {
        String queueName = queue;
        // 是否自动ack，如果不自动ack，需要使用channel.ack、channel.nack、channel.basicReject 进行消息应答
        String autoAck = null;
        int index = queue.indexOf(":");
        if (index > -1) {
          queueName = queue.trim().substring(0, index);
          autoAck = queue.trim().substring(index + 1);
        }
        try {
          // 创建队列,durable表示队列是否持久化，持久化队列在rabbitmq服务重启后还在
          channel.queueDeclare(queueName, true, false, false, null);
          if (StringUtil.equals("false", autoAck)) {
            // 设置队列是否自动ack，默认自动应答，
            channel.basicConsume(queueName, false, new DefaultConsumer(channel));
          }
          StringBuilder sb = new StringBuilder("queueDeclare成功:").append(queueName);
          if (StringUtil.isNotBlank(autoAck)) {
            sb.append(", autoAck = ").append(autoAck);
          }
          log.info(sb.toString());
        } catch (IOException e) {
          log.error("定义queue失败[{}],异常：{}", queueName, e.getMessage());
        }
      });
    }
  }

  /**
   * 为队列绑定交换机topic
   * 
   * @param connectionFactory
   * @param exchange
   * @param rotuing
   */
  private void bindExchange(CachingConnectionFactory connectionFactory, String exchange,
      String rotuing) {
    Channel channel = channel(connectionFactory);
    if (null == channel) {
      log.error("channel为空，bindExchange失败");
      return;
    }
    if (StringUtil.isAllNotBlank(exchange, rotuing)) {
      Stream.of(rotuing.trim().split("&")).filter(StringUtil::isNotBlank).forEach(rotuingName -> {
        String[] ss = rotuingName.trim().split(":");
        if (null != ss && ss.length > 1) {
          String routingKey = ss[0].trim();
          String routingValues = ss[1].trim();
          if (StringUtil.isNotBlank(routingKey) && CollectionUtil.isNotBlank(routingValues)) {
            Stream.of(routingValues.trim().split(",")).filter(StringUtil::isNotBlank)
                .forEach(queue -> {
                  try {
                    // 为队列绑定交换机topic和bindKey
                    channel.queueBind(queue.trim(), exchange, routingKey);
                    log.info(
                        "#####################      绑定topic成功: topicName={} & queue= {} & routingKey= {}",
                        exchange, queue.trim(), routingKey);
                  } catch (IOException e) {
                    log.error("bindExchange失败routing[{}],异常: {}", rotuing, e.getMessage());
                  }
                });
          } else {
            log.error("routing 格式不正确，rotuing={}", rotuingName);
          }
        } else {
          log.error("routing 格式不正确，rotuing={}", rotuingName);
        }
      });
    }
  }

  /**
   * 获得channel,并设置同一时间每次发给一个消息给一个worker
   * 
   * @param connectionFactory
   * @return
   */
  private Channel channel(CachingConnectionFactory connectionFactory) {
    Channel channel = null;
    try {
      Connection connection = connectionFactory.getRabbitConnectionFactory().newConnection();
      channel = connection.createChannel();
      // 设置同一时间每次发给一个消息给一个worker
      channel.basicQos(1, false);
    } catch (IOException e) {
      log.error("createChannel失败，异常：{}", e.getMessage());
    } catch (TimeoutException e) {
      log.error("createChannel超时，异常：{}", e.getMessage());
    }
    return channel;
  }
}
